package com.eric.demo.snowflake;

import java.util.List;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.dianping.lion.Environment;
import com.dianping.lion.client.LionException;

/**
 * @author guangyin.zhang
 * @since 2019/6/11
 */
public class ZkClient {

    private static Logger logger = LoggerFactory.getLogger(ZkClient.class);

    private static volatile ZkClient instance;

    private String address;

    private CuratorFramework curatorClient;

    private volatile boolean isConnected = false;

    private static final String CHARSET = "UTF-8";

    /**
     * 如果已经使用SpringConfig配置为Spring Bean，或者用getInstance(String address)成功调用过一次，
     * 则用此方法可以成功返回ConfigCache，否则必须使用getInstance(String address)获取ConfigCache
     *
     * @return
     * @throws LionException
     */
    public static ZkClient getInstance() throws LionException {
        if (instance == null) {
            instance = getInstance(Environment.getZKAddress());
        }
        return instance;
    }

    /**
     * @param address
     *           ip地址和端口，如127.0.0.1：2181，192.168.8.21：2181
     * @return
     * @throws LionException
     */
    public static ZkClient getInstance(String address) throws LionException {
        if (instance == null) {
            synchronized (ZkClient.class) {
                if (instance == null) {
                    try {
                        ZkClient cc = new ZkClient();
                        cc.setAddress(address);
                        cc.init();
                        instance = cc;
                    } catch (Exception e) {
                        logger.error("Failed to initialize ConfigCache", e);
                        throw new LionException(e);
                    }
                }
            }
        } else {
            if (!address.equals(instance.getAddress())) {
                throw new LionException("ConfigCache has been initialized with address " + address);
            }
        }
        return instance;
    }

    private void init() throws Exception {
        curatorClient = CuratorFrameworkFactory.newClient(address, 60*1000, 30*1000,
            new RetryNTimes(Integer.MAX_VALUE, 1000));

        curatorClient.getConnectionStateListenable().addListener(new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                logger.info("lion zookeeper state changed to {}", newState);
                if (newState == ConnectionState.CONNECTED) {
                    isConnected = true;
                } else if (newState == ConnectionState.RECONNECTED) {
                    isConnected = true;
                    try {
//                        Cat.logEvent("lion", "zookeeper:reconnected", Message.SUCCESS, address);
//                        syncConfig();
                    } catch(Exception e) {
                        logger.error("failed to watch all lion key", e);
                    }
                } else {
                    isConnected = false;
                }
            }
        });


        curatorClient.start();

    }

    public void create(String path, String data) throws Exception {
        try {
            curatorClient.create().creatingParentsIfNeeded().forPath(path, data.getBytes(CHARSET));
        } catch (Exception ex) {
            logger.error("Failed to create path " + path + ":" + data, ex);
            throw ex;
        }
    }

    public void delete(String path) throws Exception {
        try {
            curatorClient.delete().forPath(path);
        } catch (Exception ex) {
            logger.error("Failed to delete path " + path, ex);
            throw ex;
        }
    }

    public String get(String path) throws Exception {
        try {
            byte[] data = curatorClient.getData().forPath(path);
            return new String(data, CHARSET);
        } catch(KeeperException.NoNodeException ex) {
            logger.warn("", ex);
            return null;
        } catch (Exception ex) {
            logger.error("Failed to get path " + path, ex);
            throw ex;
        }
    }

    public void set(String path, String data) throws Exception {
        try {
            curatorClient.setData().forPath(path, data.getBytes(CHARSET));
        } catch (Exception ex) {
            logger.error("Failed to set path " + path + ":" + data, ex);
            throw ex;
        }
    }

    public void createOrSet(String path, String data) throws Exception {
        if(exists(path)) {
            set(path, data);
        } else {
            create(path, data);
        }
    }

    public List<String> getChildren(String path) throws Exception {
        try {
            List<String> children = curatorClient.getChildren().forPath(path);
            return children;
        } catch (Exception ex) {
            logger.error("Failed to get children of " + path, ex);
            throw ex;
        }
    }

    public boolean exists(String path) throws Exception {
        try {
            Stat stat = curatorClient.checkExists().forPath(path);
            return stat != null;
        } catch (Exception ex) {
            logger.error("Failed to check path exists " + path, ex);
            throw ex;
        }
    }

    public String getAddress() {
        return address;
    }

    public void setAddress(String address) {
        this.address = address;
    }
}
